package com.atguigu.realtime.app.func;


import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.common.EDUConfig;
import com.atguigu.realtime.utils.DruidPhoenixDSUtil;
import com.atguigu.realtime.utils.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;

public class MyPhoenixSink extends RichSinkFunction<JSONObject> {
    private DruidDataSource druidDataSource;
    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = DruidPhoenixDSUtil.getDataSource();
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        if (value.getString("type").equals("update")){

        }
        value.remove("type");
        String sql = createUpsertSql(value);
        DruidPooledConnection connection = null;
        try {
            connection = druidDataSource.getConnection();
            PhoenixUtil.executeSql(sql, connection);
        }catch (SQLException e) {
            e.printStackTrace();
            System.out.println("连接获取异常");
        }
    }

    private String createUpsertSql(JSONObject value) {
        String table = value.getString("table");
        value.remove("sink_table");
        Set<String> keys = value.keySet();
        Collection<Object> values = value.values();
        StringBuilder sql = new StringBuilder();
        sql.append("upsert into ")
                .append(EDUConfig.HBASE_SCHEMA)
                .append(".")
                .append(table)
                .append("(")
                .append(StringUtils.join(keys,","))
                .append(") values ('")
                .append(StringUtils.join(value,"','"))
                .append("')");
        return sql.toString();
    }
}
